package kafkaCase

import (
	"context"
	"errors"
	"io"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
)

func KafkaConsumer() {
	// Kafka 配置
	brokers := []string{"192.168.3.135:30092"}    // Kafka broker 地址
	topic := "dosec_asset_container_and_honeypot" // Kafka 主题
	groupID := "assets-dataa-engine"              // 消费者组 ID
	username := "admin"                           // SASL 用户名
	password := "36UdcuHtpZSfRj"                  // SASL 密码

	// 配置SASL_PLAINTEXT认证
	saslMechanism := plain.Mechanism{
		Username: username,
		Password: password,
	}

	// 创建 Kafka Reader
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  brokers,
		Topic:    topic,
		GroupID:  groupID,
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
		Dialer: &kafka.Dialer{
			Timeout:       10 * time.Second,
			DualStack:     true,
			SASLMechanism: saslMechanism,
		},
	})
	defer reader.Close()

	log.Println("Kafka 消费者已启动，等待消息...")

	// 消费消息
	for {
		msg, err := reader.ReadMessage(context.Background())
		if err != nil {
			if errors.Is(err, io.EOF) {
				log.Println("Kafka broker关闭连接，正在重试...")
				time.Sleep(5 * time.Second) // 等待5秒后重试
				continue
			}
			log.Printf("读取消息时出错: %v\n", err)
			break
		}

		log.Printf("收到消息: 主题=%s, 分区=%d, 偏移量=%d, 键=%s, 值=%s\n",
			msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
	}
}
